package sg20.task2;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;


public class TaskTwoReducer extends Reducer<LongWritable, Text, Text, Text> {

    @Override
    protected void reduce(LongWritable key, Iterable<Text> values, Reducer<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {

        String universityString = "";
        ArrayList<String> vals = new ArrayList<>();

        for (Text value : values) {
            String val = value.toString();
            if (val.startsWith("$#")) {
                // "$#..." is divided into "$" "..."
                String[] universityStrings = val.split("#");
                universityString = universityStrings[1];
            } else {
                vals.add(val);
            }
        }

        for (String val : vals) {
            context.write(new Text(val + universityString), new Text(""));
        }

    }
}
